package day05;

import data.ClickSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.util.Collector;
import po.Event;

/**
 * @Description: connect来联结两条流
 *                  1. 只能连结两条流
 *                  2. 两条流中元素类型可以不同
 * @Author: ZYX
 * @Date: 2022/2/15 10:52
 * @Version: 1.0
 */
public class Demo09 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> clickStream = env.addSource(new ClickSource());
        DataStreamSource<String> queryStream = env.socketTextStream("localhost", 9999);

        clickStream.keyBy(obj -> obj.user)
                .connect(queryStream.broadcast())
                .flatMap(new CoFlatMapFunction<Event, String, Event>() {
                    private String query = "";
                    @Override
                    public void flatMap1(Event event, Collector<Event> collector) throws Exception {
                        if (event.url.equalsIgnoreCase(query)) {
                            collector.collect(event);
                        }
                    }

                    @Override
                    public void flatMap2(String value, Collector<Event> collector) throws Exception {
                        query = value;
                    }
                }).print();

        env.execute();

    }

}
